package com.tang.state;

import com.tang.bean.WaterSensor;
import com.tang.functions.WaterSensorMapFunction;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.scala.typeutils.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.Optional;

/**
 * 检测每种传感器的水位值，如果连续的两个水位值差超过10，就输出报警。
 *
 * @author tang
 * @since 2023/7/11 11:00
 */
public class KeyedValueStateDemo {

    public static void main(String[] args) throws Exception {
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(1);
        
        OutputTag<WaterSensor> warningTag = new OutputTag<>("out", Types.POJO(WaterSensor.class));
        SingleOutputStreamOperator<WaterSensor> process = env.socketTextStream("192.168.70.141", 7777)
                .map(new WaterSensorMapFunction())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((element, recordTimestamp) -> element.getTs() * 1000L)
                ).keyBy(WaterSensor::getId)
                .process(new KeyedProcessFunction<String, WaterSensor, WaterSensor>() {

                    // 1. 定义一个ValueState，事实证明，这个ValueState和Integer他娘的是一个作用的，所以有什么卵用呢？
                    private ValueState<Integer> lastVcState;
                    private Integer i;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        // 2. 在Open方法中初始化状态
                        lastVcState = getRuntimeContext().getState(new ValueStateDescriptor<>("lastVcState", Types.INT()));
                        i = null;
                    }

                    @Override
                    public void processElement(WaterSensor waterSensor,
                                               KeyedProcessFunction<String, WaterSensor, WaterSensor>.Context ctx,
                                               Collector<WaterSensor> out) throws Exception {
                        // 3. 进行判断
                        Integer lastVc = Optional.ofNullable(lastVcState.value()).orElse(0);
                        Integer vc = waterSensor.getVc();
                        // Math.abs是啥? 取绝对值
                        if (Math.abs(vc - lastVc) > 10) {
                            System.out.println(i);
                            System.out.println("传感器=" + waterSensor.getId() + "==>当前水位值=" + vc + ",与上一条水位值=" + lastVc + ",相差超过10！！！！");
                            ctx.output(warningTag, waterSensor);
                        }
                        out.collect(waterSensor);
                        lastVcState.update(vc);
                        i = vc;
                    }
                    
                });
        
        // 4. 测输出流，输出告警
        process.print("主流");
        process.getSideOutput(warningTag).printToErr("告警流");
        
        env.execute();
    }

}
